/***************************************************************************
 * Copyright 2012 TXT e-solutions SpA
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 * 
 * This work was performed within the IoT_at_Work Project
 * and partially funded by the European Commission's
 * 7th Framework Programme under the research area ICT-2009.1.3
 * Internet of Things and enterprise environments.
 *
 * Authors:
 *      Salvatore Piccione (TXT e-solutions SpA)
 *      Carbone Matteo
 *
 * Contributors:
 *        Domenico Rotondi (TXT e-solutions SpA)
 **************************************************************************/
package it.txt.ens.authorisationService.util;

import it.txt.access.capability.pdp.client.PDPClient;
import it.txt.access.capability.pdp.client.PDPClientConfiguration;
import it.txt.access.capability.pdp.client.PDPClientFactory;
import it.txt.ens.authorisationService.amqp.ManagedBroker;
import it.txt.ens.authorisationService.amqp.ManagedNamespace;
import it.txt.ens.authorisationService.amqp.ManagedVirtualHost;
import it.txt.ens.core.ENSAuthzServiceConnectionParameters;
import it.txt.ens.core.KeystoreParameters;
import it.txt.ens.core.X509CertificateRetrievalParameters;
import it.txt.ens.core.factory.ENSResourceFactory;
import it.txt.ens.core.util.AMQPFactory;
import it.txt.ens.namespace.NamespaceInquirerConfiguration;
import it.txt.ens.namespace.NamespaceInquirerFactory;
import it.txt.ens.schema.request.factory.ENSAuthorisationRequestFactory;
import it.txt.ens.schema.response.factory.ENSAuthorisationResponseFactory;
import it.txt.rabbitmq.management.rest.factory.PermissionServiceFactory;
import it.txt.rabbitmq.management.rest.factory.UserServiceFactory;
import it.txt.rabbitmq.management.rest.resources.Queue;
import it.txt.rabbitmq.management.rest.services.QueueService;

import java.io.IOException;
import java.net.URISyntaxException;
import java.text.MessageFormat;
import java.util.Locale;
import java.util.ResourceBundle;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

import com.orientechnologies.orient.client.remote.OEngineRemote;
import com.orientechnologies.orient.core.Orient;
import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.AlreadyClosedException;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;
import com.rabbitmq.client.ShutdownListener;
import com.rabbitmq.client.ShutdownSignalException;

/**
 * This class defines a subscriber listening to access requests (actually RPC-over-AMQP messages).
 * 
 * @author Carbone Matteo
 * @author Salvatore Piccione (TXT e-solutions SpA - salvatore.piccione AT network.txtgroup.com)
 */
public class AccessRequestListener implements Runnable, ShutdownListener {
    private static final Logger LOGGER = Logger.getLogger(AccessRequestListener.class.getName());
    private static final ResourceBundle MESSAGES = ResourceBundle.getBundle("message-bundles/" + 
            AccessRequestListener.class.getSimpleName(), Locale.ROOT);
    
//    private RpcServer server;
    private AccessRequestConsumer accessRequestConsumer;
	private ENSAuthorisationResponseFactory ensResponseFactory;
//    private volatile boolean isMarkedAsClosed;
	private ENSAuthorisationRequestFactory ensRequestFactory;
	private ENSAuthzServiceConnectionParameters connParams;
	private PDPClientFactory pdpClientFactory;
    private PDPClientConfiguration pdpClientConfiguration;
	private NamespaceInquirerFactory namespaceInquirerFactory;
	private NamespaceInquirerConfiguration namespaceInquirerConfig;
	private UserServiceFactory userMgmtSF;
	private PermissionServiceFactory permissionMgmtSF;
	private KeystoreParameters keystoreParams;
	private X509CertificateRetrievalParameters certParams;
    private String consumerTag;
    private int threadPoolSize;
    private Connection connection;
    private Channel listeningChannel;
    private QueueService queueMgmt;
    private ENSResourceFactory resourceFactory;
    private volatile boolean isClosed;
    
    public AccessRequestListener (ENSAuthzServiceConnectionParameters connParams,
            PDPClientFactory pdpClientFactory,
            PDPClientConfiguration pdpClientConfiguration,
            NamespaceInquirerFactory namespaceInquirerFactory,
            NamespaceInquirerConfiguration namespaceInquirerConfig,
            UserServiceFactory userMgmtSF, PermissionServiceFactory permissionMgmtSF,
            QueueService queueService,
            ENSAuthorisationResponseFactory responseFactory,
            ENSAuthorisationRequestFactory requestFactory, KeystoreParameters keystoreParams,
            X509CertificateRetrievalParameters certParams,
            ENSResourceFactory resourceFactory,
            ConcurrentMap<String,ManagedVirtualHost> virtualHostMap,
            ConcurrentMap<String,ManagedNamespace> namespaceMap, ConcurrentMap<String,ManagedBroker> brokerMap,
            int threadPoolSize) {
        this.connParams = connParams;
        this.pdpClientFactory = pdpClientFactory;
        this.pdpClientConfiguration = pdpClientConfiguration;
        this.namespaceInquirerFactory = namespaceInquirerFactory;
        this.namespaceInquirerConfig = namespaceInquirerConfig;
        this.userMgmtSF = userMgmtSF;
        this.permissionMgmtSF = permissionMgmtSF;
        this.ensRequestFactory = requestFactory;
        this.ensResponseFactory = responseFactory;
        this.keystoreParams = keystoreParams;
        this.certParams = certParams;
        this.queueMgmt = queueService;
        this.threadPoolSize = threadPoolSize;
        AccessRequestEvaluator.namespaces = namespaceMap;
        AccessRequestEvaluator.virtualHosts = virtualHostMap;
        AccessRequestEvaluator.brokers = brokerMap;
        this.resourceFactory = resourceFactory;
        this.isClosed = true;
        
        Orient.instance().registerEngine(new OEngineRemote());
    }

//	public Channel getChannel() {
//		return channel;
//	}
//
//	public void setChannel(Channel c) {
//		channel = c;
//	}

//	public void run() {
//		try {
//			server = new StringRpcServer(channel, jsonAccessVhost.get(PropertiesKeys.VIRTUAL_HOST_QUEUE).getAsString()) {
//
//				public String handleStringCall(String request) {
//					// System.out.println("request received: " + request);
//					//declare the instance in charge of evaluating the request
//					final AccessRequestEvaluator r = new AccessRequestEvaluator(mapProperties, jsonAccessVhost, ensResponseFactory, 
//							ensRequestFactory,
//							pdpClient,
//							namespaceInquirer,
//							rabbitMQUserService,
//							rabbitMQPermissionService);
//					
//					String ans = null;
//					try {
//						ans = r.evaluate(this.getChannel(), request);
//					} catch (IOException e) {
//						e.printStackTrace();
//					}
//					return ans;
//				}
//			};
//
//			server.mainloop();
//		}
//		catch (Throwable t) {
//		    t.printStackTrace(System.err);
//		}
//
//		 try {
//			 Thread.sleep(10);
//		 }
//		 catch (InterruptedException exc) {
//			 System.err.println(exc);
//		 }
//	}
	public void start () {
        String queueName = connParams.getDestinationName();
        String vHost = connParams.getVirtualHost();
        String brokerHost = connParams.getBrokerHost();
        int brokerPort = connParams.getBrokerPort();
        try {
            connection = AMQPFactory.createConnection(connParams);
            isClosed = false;
            connection.addShutdownListener(this);
            
            listeningChannel = connection.createChannel();
            if (queueMgmt.getQueue(vHost, queueName) == null)
                queueMgmt.putQueue(new Queue(queueName, vHost, false, false, null));
        } catch (Exception e) {
            LOGGER.log(Level.SEVERE, MessageFormat.format(MESSAGES.getString("queueCreationFailure"), 
                    queueName, vHost, brokerHost + ":" + brokerPort), e);
            return;
        }

	    try {
	        //set the ThreadContext
//	        original = Thread.currentThread().getContextClassLoader();
//	        Thread.currentThread().setContextClassLoader(AMQChannel.class.getClassLoader());
	        //create the queue
            accessRequestConsumer = new AccessRequestConsumer(listeningChannel,threadPoolSize);
            
            this.consumerTag = listeningChannel.basicConsume(queueName, accessRequestConsumer);
            if (LOGGER.isLoggable(Level.INFO)) {
                LOGGER.log(Level.INFO, MessageFormat.format(MESSAGES.getString("listenerStarted"), 
                        connParams.getDestinationName(), consumerTag, connParams.getBrokerHost(),
                        connParams.getBrokerPort(), connParams.getVirtualHost()));
            }
        } catch (IOException e) {
            LOGGER.log(Level.SEVERE, MESSAGES.getString("consumerRegistrationFailure"),e);
        }
	}
	
	public void run() {
	    start();
    }

    public void stop() {
        if (!this.isClosed) {
            try {
                if (this.consumerTag != null && this.accessRequestConsumer.isRunning()) {
                    
                    try {
                        if (LOGGER.isLoggable(Level.FINEST))
                            LOGGER.log(Level.FINEST, MessageFormat.format(MESSAGES.getString(
                                    "consumerCancellingOnListeningChannel"),consumerTag));
                        listeningChannel.basicCancel(consumerTag);
                    } catch (AlreadyClosedException channelClosedExp) {
                        try {
                            if (LOGGER.isLoggable(Level.FINEST))
                                LOGGER.log(Level.FINEST, MessageFormat.format(MESSAGES.getString(
                                        "consumerCancellingOnNewChannel"),consumerTag));
                            listeningChannel = connection.createChannel();
                        } catch (AlreadyClosedException connectionClosedExp) {
                            if (LOGGER.isLoggable(Level.FINEST))
                                LOGGER.log(Level.FINEST, MessageFormat.format(MESSAGES.getString(
                                        "consumerCancellingOnNewConnection"),consumerTag));
                            connection = AMQPFactory.createConnection(connParams);
                            listeningChannel = connection.createChannel();
                        }
                        listeningChannel.basicCancel(consumerTag);
                    }
                }
            } catch (IOException e) {
                if (LOGGER.isLoggable(Level.WARNING))
                    LOGGER.log(Level.WARNING, MessageFormat.format(MESSAGES.getString(
                            "consumerCancellingError"),consumerTag), e);
                if (LOGGER.isLoggable(Level.FINEST))
                    LOGGER.log(Level.FINEST, MESSAGES.getString("manualConsumerShutdown"),consumerTag);
                accessRequestConsumer.stop();
            } finally {
                try {
                    isClosed = true;
                    if (connection != null)
                        connection.close();
                }
                catch (ShutdownSignalException  e) {}
                catch (IOException e) {}
            }
            LOGGER.log(Level.INFO, MESSAGES.getString("listenerOrderlyClosed"));
        }
    }
    
    private class AccessRequestConsumer extends DefaultConsumer {
        /* (non-Javadoc)
         * @see com.rabbitmq.client.DefaultConsumer#handleShutdownSignal(java.lang.String, com.rabbitmq.client.ShutdownSignalException)
         */
        @Override
        public void handleShutdownSignal(String consumerTag,
                ShutdownSignalException sig) {
//            if (isRunning())
                stop();
        }
        
        public boolean isRunning () {
//            return !threadPool.isShutdown();
            return false;
        }

        private ExecutorService threadPool;
//        private Set<Thread> threads;
        private final long TIMEOUT = 5;
        private final TimeUnit TIMEOUT_UOM = TimeUnit.SECONDS;
        private Connection connection;
//        private boolean closed;
        /**
         * @param channel
         * @throws IOException 
         */
        public AccessRequestConsumer(Channel channel, int threadPoolSize) throws IOException {
            super(channel);
            this.connection = AMQPFactory.createConnection(connParams);
            if (LOGGER.isLoggable(Level.FINE)) {
                LOGGER.log(Level.FINE, MessageFormat.format(MESSAGES.getString("consumerStarted"),threadPoolSize));
            }
//            threads = new HashSet<Thread>();
            this.threadPool = Executors.newFixedThreadPool(threadPoolSize);            
        }
        /* (non-Javadoc)
         * @see com.rabbitmq.client.DefaultConsumer#handleCancelOk(java.lang.String)
         */
        @Override
        public void handleCancelOk(String consumerTag) {
            stop();
        }
        /* (non-Javadoc)
         * @see com.rabbitmq.client.DefaultConsumer#handleDelivery(java.lang.String, com.rabbitmq.client.Envelope, com.rabbitmq.client.AMQP.BasicProperties, byte[])
         */
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope,
                BasicProperties properties, byte[] body) throws IOException {
            PDPClient pdpClient;
            try {
                pdpClient = pdpClientFactory.create(pdpClientConfiguration);
            } catch (URISyntaxException e) {
                LOGGER.log(Level.SEVERE, MESSAGES.getString("pdpClientObjectCreation"), e);
                getChannel().basicAck(envelope.getDeliveryTag(), false);
                return;
            }
            AccessRequestEvaluator evaluator = new AccessRequestEvaluator(
                    ensResponseFactory,
                    ensRequestFactory,
                    keystoreParams,
                    certParams,
                    resourceFactory,
                    pdpClient,
                    Boolean.parseBoolean(PropertiesKeys.NAMESPACE_CHECK_FLAG) ?
                        namespaceInquirerFactory.create(namespaceInquirerConfig): null,
                    userMgmtSF,
                    permissionMgmtSF,
                    body,
                    connection,
                    properties);
            threadPool.submit(evaluator);
            getChannel().basicAck(envelope.getDeliveryTag(), false);
            if (LOGGER.isLoggable(Level.FINER))
                LOGGER.finer(MESSAGES.getString("receivedRequest"));
//            evaluator.run();
        }
        
        public void stop () {
            this.threadPool.shutdown();
            try {
                this.threadPool.awaitTermination(TIMEOUT,TIMEOUT_UOM);
                if (LOGGER.isLoggable(Level.INFO))
                    LOGGER.info(MessageFormat.format(MESSAGES.getString("consumerOrderlyShutdown"),consumerTag));
//                Thread.currentThread().setContextClassLoader(original);
            } catch (InterruptedException e) {
                LOGGER.log(Level.WARNING,MESSAGES.getString("dirtyConsumerPoolShutdown"),e);
            }
        }
    }
    /* (non-Javadoc)
     * @see com.rabbitmq.client.ShutdownListener#shutdownCompleted(com.rabbitmq.client.ShutdownSignalException)
     */
    @Override
    public void shutdownCompleted(ShutdownSignalException cause) {
        if (!isClosed) {
            LOGGER.log(Level.SEVERE, MessageFormat.format(MESSAGES.getString("connectionShutdown"),
                    connParams.getBrokerHost() + ":" + connParams.getBrokerPort()), cause);
            if (LOGGER.isLoggable(Level.FINEST))
                LOGGER.log(Level.FINEST, MESSAGES.getString("manualConsumerShutdown"),consumerTag);
            accessRequestConsumer.stop();
            isClosed = true;
        }
    }
}
